package com.lagou

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}

object WorkFour {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getCanonicalName}")
      .config("spark.sql.crossJoin.enabled", "true")
      .master("local[*]")
      .getOrCreate()

    // 读取鸢尾花样本数据
    val df1 = spark.read.format("csv")
      .option("inferSchema", "true")
      .option("header", "true")
      .load("data/source4/Iris.csv")

    // 读取待测试数据
    val df2 = spark.read.format("csv")
      .option("inferSchema", "true")
      .option("header", "true")
      .load("data/source4/test.csv")

    df1.createOrReplaceTempView("sample")
    df2.createOrReplaceTempView("test")

    // 计算样品与测试品之间的欧氏距离
    val distanceDf: DataFrame = spark.sql(
      """
        |select test.id as testId, sample.id as sampleId, sample.species,
        | sqrt(power(test.x-sample.x, 2) + power(test.y-sample.y, 2) + power(test.m-sample.m, 2) + power(test.n-sample.n, 2)) as distance
        |from test, sample
        |""".stripMargin)

    distanceDf.createOrReplaceTempView("distanceTab")

    // 取出与测试数据距离最近的九个样品
    val topDf: DataFrame = spark.sql(
      """
        |select * from (
        | select testId, sampleId, species, distance,
        |   row_number() over (partition by testId order by distance asc) rank
        | from distanceTab ) tmp
        |where rank <= 9
        |""".stripMargin)

    topDf.createOrReplaceTempView("topTab")

    // 在最近九个选出个数最多的品种
    spark.sql(
      """
        |with tmp as (
        | select testId, species, count(*) as speciesCount
        | from topTab
        | group by testId, species
        |)
        |select testId, species from(
        | select testId, species, speciesCount,
        |   row_number() over (partition by testId order by speciesCount desc) rank
        | from tmp ) tmp2
        |where rank = 1
        |""".stripMargin).show()


    spark.close()
  }
}
